package dataStream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc union 可以将多个相同类型的流进行合并
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class UnionFunc {
    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        List<Person> data2 = new ArrayList<>();
        data2.add( new Person("man", 30));
        data2.add( new Person("man", 48));
        data2.add( new Person("man", 26));
        data2.add( new Person("woman", 21));
        data2.add( new Person("woman", 18));
        data2.add( new Person("woman", 16));
        data2.add( new Person("woman", 15));

        List<Person> data3 = new ArrayList<>();
        data3.add( new Person("man", 30));
        data3.add( new Person("man", 48));
        data3.add( new Person("man", 26));
        data3.add( new Person("woman", 21));
        data3.add( new Person("woman", 18));
        data3.add( new Person("woman", 16));
        data3.add( new Person("woman", 15));


        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream1 = env.fromCollection(data);
        DataStream<Person> dataStream2 = env.fromCollection(data2);
        DataStream<Person> dataStream3 = env.fromCollection(data3);

        SingleOutputStreamOperator<Object> singleOutputStreamOperator = dataStream1.union(dataStream2,dataStream3).map(new MapFunction<Person, Object>() {
            @Override
            public Object map(Person value) throws Exception {
                return value.getGender();
            }
        });
        singleOutputStreamOperator.print();
        env.execute();
    }
}
